package com.shujia.flink.sink;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Demo3BatchSInk {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //修改处理模式
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);

        //构建file source
        FileSource<String> fileSource = FileSource
                .forRecordStreamFormat(new TextLineInputFormat(), new Path("data/students/csv"))
                .build();

        //使用file source
        DataStream<String> studentDS = env
                .fromSource(fileSource, WatermarkStrategy.noWatermarks(), "fileSource");
        /*
         * 统计班级的人数
         */
        DataStream<Tuple2<String, Integer>> kvDS = studentDS
                //取出班级
                .map(stu -> {
                    String clazz = stu.split("\\|")[4];
                    return Tuple2.of(clazz, 1);
                })
                //显示指定返回类型
                .returns(Types.TUPLE(Types.STRING, Types.INT));

        //统计班级的人数
        DataStream<Tuple2<String, Integer>> clazzNumDS = kvDS.keyBy(kv -> kv.f0).sum(1);

        //整理数据格式
        DataStream<String> resultDS = clazzNumDS.map(kv -> {
            String clazz = kv.f0;
            Integer num = kv.f1;
            return clazz + "," + num;
        });

        //将计算结果保存到文件
        FileSink<String> fileSink = FileSink
                .forRowFormat(new Path("data/clazz_num"), new SimpleStringEncoder<String>("UTF-8"))
                .build();

        //保存结果文件
        resultDS.sinkTo(fileSink);


        env.execute();
    }
}
